package com.gugee.ins.data.analysis.listener;

import com.gugee.ins.data.analysis.config.InsAnalysisConfig;
import com.gugee.ins.data.analysis.service.InsAnalysisCategoryService;
import com.gugee.ins.data.analysis.service.InsAnalysisProductCategoryService;
import com.gugee.ins.data.common.consumer.AbstractGugeeConsumerHandler;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.common.msg.InsProductMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;


/**
 * InsAnalysisCategory Listener
 *
 * @author Huangxin
 */
@Component
@Slf4j
public class InsAnalysisCategoryListener {

    /**
     * InsAnalysisCategory
     */
    @Autowired
    private InsAnalysisConfig insAnalysisConfig;

    /**
     * InsAnalysisCategory
     */
    @Autowired
    private InsAnalysisCategoryService insAnalysisCategoryService;

    /**
     * insAnalysisProductCategoryService
     */
    @Autowired
    private InsAnalysisProductCategoryService insAnalysisProductCategoryService;

    private InsAnalysisCategoryHandler analysisHandler = new InsAnalysisCategoryHandler();

    private InsAnalysisProductCategoryHandler analysisProductCategoryHandler = new InsAnalysisProductCategoryHandler();

    /**
     * 监听任务,完成播主分类分析
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisCategory.id}",
            topics = {"${kafka.topics.insAnalysisCategory.online}"},
            containerFactory = "factoryInsAnalysisCategoryAck")
    public void insAnalysisCategory(List<InsBloggerMessage> messages, Acknowledgment ack) {
        analysisHandler.handle("InsAnalysisCategory",messages, insAnalysisConfig.getAcceptInsAnalysisCategoryMsgTimeMinRange(),ack);
    }

    /**
     * 监听任务,完成商品分类分析
     * @param messages
     * @param ack
     */
    @KafkaListener(id="${kafka.consumer.group.insAnalysisProductCategory.id}",
            topics = {"${kafka.topics.insAnalysisProductCategory.online}"},
            containerFactory = "factoryInsAnalysisProductCategoryAck")
    public void insAnalysisProductCategory(List<InsProductMessage> messages, Acknowledgment ack) {
        analysisProductCategoryHandler.handle("InsAnalysisProductCategory",messages, insAnalysisConfig.getAcceptInsAnalysisProductCategoryMsgTimeMinRange(),ack);
    }

    class InsAnalysisCategoryHandler extends AbstractGugeeConsumerHandler<InsBloggerMessage>{
        @Override
        public void dealMsg(List<InsBloggerMessage> messages) {
            insAnalysisCategoryService.handleAnalysisCategory(messages);
        }
    }

    class InsAnalysisProductCategoryHandler extends AbstractGugeeConsumerHandler<InsProductMessage>{
        @Override
        public void dealMsg(List<InsProductMessage> messages) {
            insAnalysisProductCategoryService.handle(messages);
        }
    }
}
